package com.nepu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.util.DateFormatUtil;
import com.nepu.gmall.realtime.util.KafkaUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 本类主要的作用是处理日志数据的，我们知道我们的日志数据是在ods层(kafka)的topic_log中的，日志这部分一共有五类
 * （1）页面日志
 * （2）启动日志
 * （3）动作日志
 * （4）曝光日志
 * （5）错误日志
 * 现在这三类日志混在一起，我们需要将这三类之日分开，分别发送到对应的topic中
 * 日志之间的关系是：
 *  1.1、启动日志和页面日志之间是冲突的。如果一条日志是页面日志，那么这天日志就不可能是启动日志
 *  1.2、错误日志和这两类日志是可以共存的
 *  1.3、我们在处理日志的时候需要将每一部分都进行拆分
 *
 *  本类对于数据处理的总统的流程是：
 *      （1）加载环境
 *      （2）设置checkpoints和StateBackEnd
 *      （3）从Kafka的topic_log中消费数据
 *      （4）对数据进行清洗，清洗掉那些不符合json格式的数据，并且将数据转化成Json格式
 *      （5）对新老用户进行校验：
 *          这个校验的原因是，前端埋点对于is_new字段的组装，使用的是缓存信息来进行判断的，如果当前登录的用户并不是
 *          新用户，他只是在一段时间之内，将该应用卸载了，清理掉了设备中的缓存，对于前端，这部分的问题是校验不出来的，
 *          所以这里要对这个字段进行校验。
 *          这个字段可能会出现以下的情况：
 *              （1）is_new = 1 : 表示的是当前用户是新用户。
 *                  如果状态中维护的时间是null，表明这个用户确实是新用户，然后将当前的时间戳存储到状态中
 *                  如果状态中维护的时间戳不是null，且这个时间戳和当前的时间戳不是一天，将is_new的值更改为0
 *                  如果状态中维护的时间戳不是null，且这个时间戳和当前的时间是一天，不做操作
 *               (2) is_new = 0 : 表示的是当前的用户是老用户
 *                  按道理说对于老用户，我们是可以不做处理的，但是因为我们的实时数仓的上线时间是要滞后于业务系统的
 *                  所以我们需要将，这些老用户的字段在状态中补全，解决的办法是，让状态中存储当前时间的前一天。
 *     （6）对数据进行拆分
 *     （7）将数据写入到对应  的topic中
 *          这里使用的是侧输出流的方式，我们将页面日志写入到主流，其他的日志写入到对应的侧输出流中
 * 测试的时候数据的整体的流向
 *  web/app -->Ngnix --> 日志服务器 --> kafka  --> BaseLogApp --> kafka
 *  mock --> f1.sh --> kafka(zookeeper) --> BaseLogApp --> BaseLogApp
 * @author chenshuaijun
 * @create 2023-02-18 15:25
 */
public class BaseLogApp {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 生产环境下是一定不会将任务的并行度设置为1的，这里具体的设置是和我们等下要读取的kafka的相应的主题的分区的个数相同
        env.setParallelism(1);
        // 设置checkpoint的信息：设置checkpoint的间隔是5分钟,并且checkpoint的级别是精确一次性
        /*env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间是10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        // 设置外部检查点。可以将检查点的元数据信息定期写入外部系统，这样当job失败时，检查点不会被清除。这样如果job失败，可以从检查点恢复job。
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置checkpoint的重启的策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)));
        // 设置两个checkpoint之间的最小的间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // 设置状态后端: 设置状态后端为内存级别
        env.setStateBackend(new HashMapStateBackend());
        // 设置checkpoint的存储的路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/checkpoint");
        // 因为我们的HDFS只有atguigu用户才能够操作，所以要将用户设置为atguigu
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/

        // TODO 1、从kafka中读取数据
        String topic = "topic_log";
        String group_id = "BaseLogApp";
        DataStreamSource<String> stringDataStreamSource = env.addSource(KafkaUtils.getKafkaConsumer(topic, group_id));
        // TODO 2、将过滤掉脏的数据,将脏数据放入到侧输出流中，所以这里使用的是processFunction
        OutputTag<String> dirtyDataTag = new OutputTag<String>("dirty-data"){};
        SingleOutputStreamOperator<JSONObject> isJsonDataStream = stringDataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    // 1、先将数据转换成Json对象
                    JSONObject jsonObject = JSON.parseObject(value);
                    // 2、将数据输出
                    out.collect(jsonObject);
                } catch (Exception e) {
                    // 如果转换失败，说明这条数据是一个脏数据，直接将数据放入到侧输出流中
                    ctx.output(dirtyDataTag, value);
                }
            }
        });

        DataStream<String> dirtyData = isJsonDataStream.getSideOutput(dirtyDataTag);
        dirtyData.print("不符合Json格式的脏数据-->" + dirtyData);

        // TODO 2、为了状态的维护，需要先将数据进行分组
        KeyedStream<JSONObject, String> jsonObjectStringKeyedStream = isJsonDataStream.keyBy(value -> value.getJSONObject("common").getString("mid"));
        // TODO 3、对is_new进行处理
        SingleOutputStreamOperator<JSONObject> isNewStream = jsonObjectStringKeyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<Long> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("last-login", Long.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                // 1、先取出对应的is_new字段
                String isNew = value.getJSONObject("common").getString("is_new");
                // 2、取出状态
                Long firstLogin = valueState.value();
                // 3、取出时间戳
                Long ts = value.getLong("ts");
                // 将当前的时间戳转换为天
                String currentDate = DateFormatUtil.toDate(ts);
                if ("1".equals(isNew)) {
                    // 如果当前的用户是新用户
                    if (firstLogin == null) {
                        // 如果当前用户确实是新用户，将当前的时间错放入到状态中
                        valueState.update(ts);
                    } else if (!currentDate.equals(DateFormatUtil.toDate(firstLogin))) {
                        // 如果这两个时间不是一天，直接将is_new改为0
                        value.getJSONObject("common").put("is_new", 0);
                    }
                } else if (firstLogin == null) {
                    // 如果当前已经是老用户了，并且状态中存储的时间是null，直接将状态更新到ts的前一天
                    valueState.update(ts - 24 * 60 * 60 * 1000L);
                }
                return value;
            }
        });

        // TODO 4、对数据进行拆解，放入到对应的流中
        // 定义对应的侧输出流的标签
        OutputTag<String> errorTag = new OutputTag<String>("error"){};
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        OutputTag<String> activateTag = new OutputTag<String>("activate"){};
        OutputTag<String> displayTag = new OutputTag<String>("display"){};

        SingleOutputStreamOperator<String> splitLogStream = isNewStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                // 1、先处理错误数据，对于错误数据，我们不做更改，直接将信息输出
                JSONObject err = value.getJSONObject("err");
                if (err != null) {
                    // 直接将整条数据输出
                    ctx.output(errorTag, value.toJSONString());
                }
                // 2、删除掉与err相关的字段
                value.remove("err");
                // 3、判断当前数据是否为启动数据
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    // 直接将整条数据输出
                    ctx.output(startTag, value.toJSONString());
                } else {
                    // 如果不是启动数据，只能是页面数据了
                    // 先取出公共的数据
                    String common = value.getString("common");
                    String pageId = value.getJSONObject("page").getString("page_id");
                    // 取出行为数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null && actions.size() > 0) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject actionObject = actions.getJSONObject(i);
                            actionObject.put("common", common);
                            actionObject.put("page_id", pageId);
                            ctx.output(activateTag, actionObject.toJSONString());
                        }
                    }
                    // 移除行为数据，即使没有也没问题
                    value.remove("actions");

                    // 取出曝光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject displayObject = displays.getJSONObject(i);
                            displayObject.put("common", common);
                            displayObject.put("page_id", pageId);
                            // 因为曝光数据中没有单独的时间戳字段，所以这里给出一个
                            displayObject.put("ts", value.getLong("ts"));
                            // 将数据写入到测输出流
                            ctx.output(displayTag, displayObject.toJSONString());
                        }
                    }
                    // 移除曝光数据
                    value.remove("displays");
                    // 将页面数据从主流中输出
                    out.collect(value.toJSONString());
                }
            }
        });

        // 取出测输出流中的数据
        DataStream<String> errorData = splitLogStream.getSideOutput(errorTag);
        DataStream<String> startData = splitLogStream.getSideOutput(startTag);
        DataStream<String> activateData = splitLogStream.getSideOutput(activateTag);
        DataStream<String> displayData = splitLogStream.getSideOutput(displayTag);

        // 所有的topic
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        splitLogStream.addSink(KafkaUtils.getFlinkKafkaProducer(page_topic));
        startData.addSink(KafkaUtils.getFlinkKafkaProducer(start_topic));
        displayData.addSink(KafkaUtils.getFlinkKafkaProducer(display_topic));
        activateData.addSink(KafkaUtils.getFlinkKafkaProducer(action_topic));
        errorData.addSink(KafkaUtils.getFlinkKafkaProducer(error_topic));

        env.execute();


    }
}
